package com.flink.api.unit5;

import com.flink.pojo.WaterSensor;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.aggregation.AggregationFunction;

/**
 * reduce_1
 *
 * @author caizhiyang
 * @since 2024-02-22
 */
public class reduce_1 {

    /**
     * reduce 规约聚合
     * @param args
     */
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.socketTextStream("192.168.226.130", 7777)
                // 转换成WaterSensor
                .map(value -> {
                    String[] datas = value.split(",");
                    return new WaterSensor(datas[0], Long.valueOf(datas[1]), Integer.valueOf(datas[2]));
                })
                //根据id分组
                .keyBy(WaterSensor::getId)
                .reduce(new AggregationFunction<WaterSensor>() {
                    @Override
                    public WaterSensor reduce(WaterSensor w1, WaterSensor w2) throws Exception {
                        int maxVc = Math.max(w1.getVc(), w2.getVc());
                        //实现max(vc)的效果  取最大值，其他字段以当前组的第一个为主
                        //value1.setVc(maxVc);
                        //实现maxBy(vc)的效果  取当前最大值的所有字段
                        if (w1.getVc() > w2.getVc()) {
                            w1.setVc(maxVc);
                            return w1;
                        } else {
                            w2.setVc(maxVc);
                            return w2;
                        }
                    }
                }).print();

        env.execute();
    }
}
